package cn.jly.bigdata.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/29 13:50
 */
object Spark15_Broadcast {

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("test-broadcast"))

    val dataRdd: RDD[(Int, Int)] = sc.makeRDD(List((1, 111), (2, 222), (3, 333)))

    val list = List((1, 1), (2, 2), (3, 3))

    // 可以使用广播变量减少数据的传输
    // 1. 构建广播向量
    val broadcast: Broadcast[List[(Int, Int)]] = sc.broadcast(list)

    val resultRdd: RDD[(Int, (Int, Any))] = dataRdd.map {
      case (key, value) => {
        var v2: Any = null

        for (elem <- broadcast.value) {
          if (key == elem._1) {
            v2 = elem._2
          }
        }

        (key, (value, v2))
      }
    }
    resultRdd.foreach(println)
  }
}
